package com.ptp.backendcertms.basiccompnents.thread;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.*;

/**
 * 异步线程池配置
 */
@Configuration
@EnableAsync
@Slf4j
@Getter
public class AsyncPoolConfig {
    @Value("${spring.async.corePoolSize:5}")
    private int corePoolSize;
    @Value("${spring.async.maxPoolSize:32}")
    private int maxPoolSize;
    @Value("${spring.async.queueCapacity:3000}")
    private int queueCapacity;
    @Value("${spring.async.keepAliveSeconds:120}")
    private int keepAliveSeconds;

    /**
     * job任务异步线程池
     */
    @Bean("asyncJobExecutor")
    public Executor getJobAsyncExecutor() {
        return buildExecutor("job-AsyncPool-");
    }

    /**
     * 事件监听异步线程池
     */
    @Bean("asyncEventExecutor")
    public Executor getEventAsyncExecutor() {
        return buildExecutor("event-AsyncPool-");
    }

    /**
     * 文件上传异步线程池
     */
    @Bean("asyncFileExecutor")
    public Executor getFileAsyncExecutor() {
        return buildExecutor("file-AsyncPool-");
    }

    /**
     * 业务处理异步线程池
     */
    @Bean("asyncBusinessExecutor")
    public Executor getBusinessAsyncExecutor() {
        return buildExecutor("business-AsyncPool-");
    }

    /**
     * 居民生产消息业务处理异步线程池
     */
    @Bean("asyncResidentProduceExecutor")
    public Executor getResidentProduceExecutor() {
        int availableProcessors = Runtime.getRuntime().availableProcessors()*60;
        int queueCapacity = availableProcessors * 1000000;
        return buildExecutor("resident-produce-", availableProcessors ,availableProcessors,queueCapacity,240);
    }

    /**
     * 居民消费消息业务处理异步线程池
     */
    @Bean("asyncResidentConsumeExecutor")
    public Executor getResidentConsumeExecutor() {
        int availableProcessors = Runtime.getRuntime().availableProcessors()*110;
        int queueCapacity = availableProcessors * 1000000;
        return buildExecutor("resident-consume-", availableProcessors ,availableProcessors,queueCapacity,300);
    }

    /**
     * 构造线程池
     */
    private Executor buildExecutor(String poolName) {
        log.info("线程池属性：CorePoolSize:" + getCorePoolSize() + "\tMaxPoolSize:" + getMaxPoolSize() + "\tQueueCapacity:" + getQueueCapacity()
                + "\tKeepAliveSeconds:" + getKeepAliveSeconds());
        // 初始化线程工厂与阻塞队列
        ThreadFactory threadFactory = new CustomizableThreadFactory(poolName);
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(getQueueCapacity());
        // 初始化线程池执行器
        MyThreadPoolExecutor executor = new MyThreadPoolExecutor(getCorePoolSize(), getMaxPoolSize(), getKeepAliveSeconds(), TimeUnit.SECONDS, workQueue, threadFactory);
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
    /**
     * 构造线程池
     */
    private Executor buildExecutor(String poolName, int corePoolSize, int maxPoolSize, int queueCapacity, int keepAliveSeconds) {
        log.info("线程池属性：CorePoolSize:" + corePoolSize + "\tMaxPoolSize:" + maxPoolSize + "\tQueueCapacity:" + queueCapacity
                + "\tKeepAliveSeconds:" + keepAliveSeconds);
        // 初始化线程工厂与阻塞队列
        ThreadFactory threadFactory = new CustomizableThreadFactory(poolName);
        BlockingQueue<Runnable> workQueue = new LinkedBlockingDeque<>(queueCapacity);
        // 初始化线程池执行器
        MyThreadPoolExecutor executor = new MyThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveSeconds, TimeUnit.SECONDS, workQueue, threadFactory);
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}
